package com.wjk.kylin.rpc.core.provider;

import com.wjk.kylin.rpc.core.config.RpcProviderConfigProperties;
import com.wjk.kylin.rpc.core.constant.Constants;
import com.wjk.kylin.rpc.core.netty.AbstractServer;
import com.wjk.kylin.rpc.core.netty.params.RpcRequest;
import com.wjk.kylin.rpc.core.netty.params.RpcResponse;
import com.wjk.kylin.rpc.core.netty.server.NettyServer;
import com.wjk.kylin.rpc.core.registry.Register;
import com.wjk.kylin.rpc.core.util.IpUtil;
import com.wjk.kylin.rpc.core.util.NetUtil;
import com.wjk.kylin.rpc.core.util.RpcException;
import com.wjk.kylin.rpc.core.util.ThrowableUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;

/**
 * provider
 * @author wjk
 */
public class RpcProviderFactory {
    private static final Logger LOGGER = LoggerFactory.getLogger(RpcProviderFactory.class);

    private RpcProviderConfigProperties properties;

    public RpcProviderConfigProperties getProperties() {
        return properties;
    }

    public void setProperties(RpcProviderConfigProperties properties) {
        this.properties = properties;
    }

    /**
     * server default port
     */
    private int port = 9527;
    public int getPort() {
        return port;
    }

    private final AbstractServer server = new NettyServer();
    private Register register;
    /**
     * init local rpc service map
     */
    private final Map<String, Object> serviceData = new HashMap<>();

    /**
     * 启动
     *
     * @throws Exception
     */
    public void start() throws Exception {
        // valid
        if (this.properties == null) {
            throw new RpcException("kylin-rpc provider config missing.");
        }
        if (this.properties.getRegistry() == null) {
            throw new RpcException("kylin-rpc provider registry type missing.");
        }
        if (this.properties.getRegistry().getType() == null) {
            throw new RpcException("kylin-rpc provider registry type missing.");
        }
        if (StringUtils.isBlank(this.properties.getRegistry().getAddress())) {
            throw new RpcException("kylin-rpc provider registry address missing.");
        }
        Class<? extends Register> serviceRegistry = this.properties.getRegistry().getType().getClazz();
        if (serviceRegistry == null) {
            throw new RpcException("kylin-rpc provider registry missing.");
        }
        register = serviceRegistry.getDeclaredConstructor().newInstance();

        if (this.properties.getSerializer() == null) {
            throw new RpcException("kylin-rpc provider serializer missing.");
        }
        if (this.properties.getCorePoolSize() < 0) {
            throw new RpcException("kylin-rpc provider thread pool must corePoolSize >=0.");
        }
        if (this.properties.getMaxPoolSize() <= 0) {
            throw new RpcException("kylin-rpc provider thread pool must maxPoolSize >0.");
        }
        if (this.properties.getMaxPoolSize() < this.properties.getCorePoolSize()) {
            throw new RpcException("kylin-rpc provider thread pool must maxPoolSize  >= corePoolSize.");
        }
        if (this.properties.getPort() <= 0) {
            throw new RpcException("kylin-rpc provider port must > 0.");
        }
        if (NetUtil.isPortUsed(this.properties.getPort())) {
            throw new RpcException("kylin-rpc provider port[" + this.properties.getPort() + "] is used.");
        }

        // server ip, for registry
        String ip = this.properties.getIp();
        this.port = this.properties.getPort();
        if (StringUtils.isBlank(ip)) {
            ip = IpUtil.getIp();
        }
        //ip:port
        String registryAddress = IpUtil.getIpPort(ip, this.port);

        // start server
        //添加启动netty服务执行的方法 主要是将接口注册到注册中心
        // serviceRegistry start
        server.setStartCallback(() -> {
            // start registry
            if (register != null) {
                register.start(this.properties.getRegistry().getAddress());
                if (serviceData.size() > 0) {
                    register.registry(serviceData.keySet(), registryAddress);
                }
            }
        });
        //添加关闭netty服务执行的方法
        server.setStopCallback(() -> {
            // stop registry
            if (register != null) {
                if (serviceData.size() > 0) {
                    register.remove(serviceData.keySet(), registryAddress);
                }
                register.stop();
                register = null;
                serviceData.clear();
            }
        });

        //启动
        server.start(this);
    }

    public void stop() throws Exception {
        // stop server
        server.stop();
    }


    // ---------------------- server invoke ----------------------

    /**
     * make service key
     *
     * @param interfaceName
     * @param version
     * @return
     */
    public static String makeServiceKey(String interfaceName, String version) {
        String serviceKey = interfaceName;
        if (StringUtils.isNotBlank(version)) {
            serviceKey += "#".concat(version.trim());
        }
        return serviceKey;
    }

    /**
     * add service
     *
     * @param interfaceName
     * @param version
     * @param serviceBean
     */
    public void addService(String interfaceName, String version, Object serviceBean) {
        String serviceKey = makeServiceKey(interfaceName, version);
        serviceData.put(serviceKey, serviceBean);

        LOGGER.info(">>>>>>>>>>> kylin-rpc provider factory add service success. serviceKey = {}, serviceBean = {}", serviceKey, serviceBean.getClass());
    }

    /**
     * invoke service
     *
     * @param rpcRequest
     * @return
     */
    public RpcResponse invokeService(RpcRequest rpcRequest) {

        //  make response
        RpcResponse rpcResponse = new RpcResponse();
        rpcResponse.setRequestId(rpcRequest.getRequestId());

        // match service bean
        String serviceKey = makeServiceKey(rpcRequest.getClassName(), rpcRequest.getVersion());
        Object serviceBean = serviceData.get(serviceKey);

        // valid
        if (serviceBean == null) {
            rpcResponse.setErrorMsg("The serviceKey[" + serviceKey + "] not found.");
            return rpcResponse;
        }

        if (System.currentTimeMillis() - rpcRequest.getCreateMillisTime() > Constants.RPC_OVER_TIME) {
            rpcResponse.setErrorMsg("The timestamp difference between admin and executor exceeds the limit.");
            return rpcResponse;
        }
        if (StringUtils.isNotBlank(this.properties.getAccessToken()) && !StringUtils.equals(this.properties.getAccessToken().trim(), rpcRequest.getAccessToken())) {
            rpcResponse.setErrorMsg("The access token[" + rpcRequest.getAccessToken() + "] is wrong.");
            return rpcResponse;
        }

        try {
            // invoke
            Class<?> serviceClass = serviceBean.getClass();
            String methodName = rpcRequest.getMethodName();
            Class<?>[] parameterTypes = rpcRequest.getParameterTypes();
            Object[] parameters = rpcRequest.getParameters();

            Method method = serviceClass.getMethod(methodName, parameterTypes);
            method.setAccessible(true);
            Object result = method.invoke(serviceBean, parameters);

            rpcResponse.setResult(result);
        } catch (Throwable t) {
            // catch error
            LOGGER.error("kylin-rpc provider invokeService error", t);
            rpcResponse.setErrorMsg(ThrowableUtil.toString(t));
        }

        return rpcResponse;
    }

}
